package com.hhf.rrd.dataset;

import org.apache.commons.io.FileUtils;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.configuration.Configuration;

import java.io.File;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.List;

/**
 * // todo
 *
 * @author huanghaifeng15
 * @date 2022/2/11 11:33
 **/
public class DistributedCacheApp {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 把对应的数据注册到分布式缓存中
        env.registerCachedFile("data/wc/wc.txt","hhf-wc");

        DataSource<String> dataSource = env.fromElements("spark", "flink", "spark", "hadoop");
        dataSource.map(new RichMapFunction<String, String>() {

            private final List<String> list = new ArrayList<>();

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                File file = getRuntimeContext().getDistributedCache().getFile("hhf-wc");
                List<String> lineList = FileUtils.readLines(file, Charset.defaultCharset());
                for (String line : lineList) {
                    list.add(line);
                    System.out.println("line --> " + line);
                }
            }

            @Override
            public String map(String value) throws Exception {
                return value;
            }
        }).print();
    }
}
